package com.csthink.mr.join.reducejoin;

import com.csthink.utils.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 需求: 读取 data/join 目录下的两个文件（dept.txt 部门表数据，emp.txt 员工表数据）
 * 实现 select e.emp_id, e.emp_name, e.dept_id, d.dept_name from emp e join dept d on emp.dept_id = dept.dept_id;
 *
 * @author <a href="mailto:csthink@icloud.com">Mars</a>
 * @since 2024-04-16 15:08
 */
public class ReduceJoinApp {

    public static void main(String[] args) throws Exception {
        String input = "data/join";
        String output = "out";

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJobName("ReduceJoinApp");

        FileUtils.deleteIfExists(conf, output);

        job.setJarByClass(ReduceJoinApp.class);

        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);

        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(EmpInfo.class);

        job.setOutputKeyClass(EmpInfo.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job, new Path(input));
        FileOutputFormat.setOutputPath(job, new Path(output));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    public static class MyMapper extends Mapper<LongWritable, Text, IntWritable, EmpInfo> {

        String tblFilePath;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            FileSplit fileSplit = (FileSplit) context.getInputSplit();
            tblFilePath = fileSplit.getPath().toString();
            System.out.println("current file name =====> " + tblFilePath);
        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] splits = value.toString().split("\t");

            if (tblFilePath.contains("emp")) {
                String empId = splits[0].trim();
                String empName = splits[1].trim();
                int deptNo = Integer.parseInt(splits[7].trim());

                EmpInfo empInfo = new EmpInfo();
                empInfo.setEmpId(empId);
                empInfo.setEmpName(empName);
                empInfo.setDeptNo(deptNo);
                // 注意：对象需要赋初始值，否则序列化时会报NPE
                empInfo.setDeptName("");
                empInfo.setFlag(1);

                context.write(new IntWritable(deptNo), empInfo);
            } else if (tblFilePath.contains("dept")) {
                int deptNo = Integer.parseInt(splits[0].trim());
                String deptName = splits[1].trim();

                EmpInfo empInfo = new EmpInfo();
                empInfo.setEmpId("");
                empInfo.setEmpName("");
                empInfo.setDeptNo(deptNo);
                empInfo.setDeptName(deptName);
                empInfo.setFlag(2);

                context.write(new IntWritable(deptNo), empInfo);
            }

        }
    }

    public static class MyReducer extends Reducer<IntWritable, EmpInfo, EmpInfo, NullWritable> {
        @Override
        protected void reduce(IntWritable key, Iterable<EmpInfo> values, Context context) throws IOException, InterruptedException {
            String deptName = "";

            List<EmpInfo> list = new ArrayList<>();

            for (EmpInfo empInfo : values) {
                if (empInfo.getFlag() == 1) {
                    EmpInfo tmpEmpInfo = new EmpInfo();
                    tmpEmpInfo.setEmpId(empInfo.getEmpId());
                    tmpEmpInfo.setEmpName(empInfo.getEmpName());
                    tmpEmpInfo.setDeptNo(empInfo.getDeptNo());
                    list.add(tmpEmpInfo);
                } else if (empInfo.getFlag() == 2) {
                    deptName = empInfo.getDeptName();
                }
            }

            for (EmpInfo empInfo : list) {
                empInfo.setDeptName(deptName);
                context.write(empInfo, NullWritable.get());
            }
        }
    }
}
